package jzh.cloud.canal.listener;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.xpand.starter.canal.annotation.CanalEventListener;
import com.xpand.starter.canal.annotation.ListenPoint;
import lombok.AllArgsConstructor;
import org.apache.rocketmq.spring.core.RocketMQTemplate;

@CanalEventListener
@AllArgsConstructor
public class UserListener {

    private static final String POSITION = "sex";

    private final RocketMQTemplate rocketMQTemplate;

    @ListenPoint(schema = "pineapple_0",table = {"p_user_0"},eventType = CanalEntry.EventType.INSERT)
    public void insert(CanalEntry.EventType eventType,CanalEntry.RowData rowData){
        System.out.println("用户数据发生变化" + rowData);

        //修改前数据
        rowData.getBeforeColumnsList().forEach(column -> {
            if (column.getName().equals(POSITION)){
                System.out.println("发送消息到mq  user_insert_queue:" + column.getValue());
                //发送消息到MQ
                rocketMQTemplate.syncSend("pineapple-user", column.getValue());
            }
        });

        //修改后数据
        for (CanalEntry.Column column : rowData.getAfterColumnsList()) {
            if (column.getName().equals(POSITION)){
                System.out.println("发送消息到mq  user_insert_queue:"+column.getValue());
                //发送消息到MQ
                rocketMQTemplate.syncSend("pineapple-user",column.getValue());
                break;
            }
        }
    }

    @ListenPoint(schema = "pineapple_0",table = {"p_goods_0"},eventType = CanalEntry.EventType.UPDATE)
    public void update(CanalEntry.EventType eventType,CanalEntry.RowData rowData){
        System.out.println("用户数据发生变化" + rowData);

        //修改前数据
        rowData.getBeforeColumnsList().forEach(column -> {
            if (column.getName().equals(POSITION)){
                System.out.println("发送消息到mq  user_insert_queue:" + column.getValue());
                //发送消息到MQ
                rocketMQTemplate.syncSend("pineapple-user", column.getValue());
            }
        });

        //修改后数据
        for (CanalEntry.Column column : rowData.getAfterColumnsList()) {
            if (column.getName().equals(POSITION)){
                System.out.println("发送消息到mq  user_insert_queue:"+column.getValue());
                //发送消息到MQ
                rocketMQTemplate.syncSend("pineapple-user",column.getValue());
                break;
            }
        }
    }


}
